package org.example.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * rdd的算子，也就是rdd的方法
 *
 *
 */
object Spark16_RDD_Operator_AggregateByKey {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)
    // key value 类型
    val rdd1 = context.makeRDD(List(
      ("a",1),("a",2),("a",3),("b",4)),2)

    /**
      * 这里需要说道特殊的两个概念
      * 之前我们再说groupbykey的时候
      * 说到了落盘和落盘前聚合的概念
      * 那么为了保证最后被shuffle后，分区内容的数据总是可以正确计算
      * 需要保证分区中他们的聚合规则，和分区之间的聚合规则要一致
      * 才能保证让计算结果正确。
      *那么这里就引申出了两个词
      *
      * 分区内 单独分区中，自己的规则
      *
      * 分区间 分区与分区之间的规则
      *
      * 在spark中，aggregateByKey可以定义
      * 分区间和分区内的规则
      */
    /**
      * 这个方法存在函数科里化
      * 第一个参数列表
      *   第一个参数，表示初始值，
      *   主要用于当碰见第一个key的时候，和value进行分区计算
      *   在这个方法中，由于(a,1)中的1，刚开始，并不知道1要和谁比大小，所以需要初始值
      *   来界定大的概念
      * 第二个参数列表
      *   第一个参数表示分区内的计算规则
      *   第二个参数表示分区间的计算规则
      */
    rdd1.aggregateByKey(
0
    )(
      (x,y)=> math.max(x,y),
      (x,y)=>x+y
    )

    // 当然你也可以分区间和分区内的计算规则一致
    rdd1.aggregateByKey(
      0
    )(
      (x,y)=> x+y,
      (x,y)=>x+y
    )
    // 简写
    rdd1.aggregateByKey(
      0
    )(_+_,_+_)
    // 更简洁
    rdd1.foldByKey(0)(_+_)

    /**
      * 附加小练习
      *
      * 希望从这里获得相同key的均值
      *
      * 也就是对于key为a的来说，是(a,3)
      * b 则是 (b,4)
      */
    val rdd2 = context.makeRDD(List(
      ("a",1),("a",2),("b",3),
      ("b",4),("b",5),("a",6)),2)

    /**
      * 从这个方法的签名上来看，我们传入的初始值类型
      * 也就是zeroValue，会影响这个方法的分组情况
      * seqOp中的U是我们的初始值，V则是rdd中的对应key的value
      * 在区间内中，则是我们的处理好的初始值类型的计算
      * def aggregateByKey[U](zeroValue: U)
      * (seqOp: (U, V) => U,combOp: (U, U) => U)
      * (implicit evidence$3: ClassTag[U]): RDD[(K, U)]
       *其实计算均值，也就是拿到key聚合的总数除以key值
      *(key聚合的value的总数，key出现的次数)
      */

    val argRdd: RDD[(String, (Int, Int))] = rdd2.aggregateByKey((0, 0))(
      (t, v) => {
        // 前者为聚合总数，后者为出现次数
        (t._1 + v, t._2 + 1)
      },
      (t1, t2) => {
        // 对区间内的结果进行区间间的计算
        (t1._1 + t2._1, t1._2 + t2._2)
      }
    )
    // 统计完毕后，进行相除
   // 但实际上，我们这个的key，也就是String，可以忽略，我们只在乎值
    val value: RDD[(String, Int)] = argRdd.mapValues {
      case (num, count) => {
        num / count
      }
    }
    // 结果计算

    context.stop()
  }
}
